package cn.ivicar.cloud.influxdb;

import java.util.*;
import java.util.concurrent.TimeUnit;

import cn.ivicar.cloud.utils.DataType;
import net.sf.json.JSONArray;
import net.sf.json.JSONObject;
import org.influxdb.InfluxDB;
import org.influxdb.InfluxDB.ConsistencyLevel;
import org.influxdb.InfluxDBFactory;
import org.influxdb.dto.BatchPoints;
import org.influxdb.dto.Point;
import org.influxdb.dto.Query;
import org.influxdb.dto.QueryResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class InfluxdbImpl implements InfluxdbIntf {
    private static transient final Logger LOGGER = LoggerFactory.getLogger(InfluxdbImpl.class);
    private static InfluxDB influxDB;

    public InfluxdbImpl(String dbUrl, String username, String password) {
        if (influxDB == null) {
            influxDB = InfluxDBFactory.connect(dbUrl, username, password);

            if (influxDB == null) {
                throw new InternalError("Influxdb初始化失败！");
            }
        }
    }

    public void write(String database, String clientId, DataType type, JSONObject data) {
        try {
            String measurement = type.toString();
            Map<String, String> tags = new HashMap<String, String>();
            tags.put("clientId", clientId);

            Map fields = (Map) data;
            Long timestamp = data.getLong("time");
            fields.remove("time");

            Point point = Point.measurement(measurement)
                    .time(timestamp, TimeUnit.MILLISECONDS).tag(tags).fields(fields).build();

            influxDB.write(database, "default", point);
        } catch (Exception e) {
            LOGGER.error(String.format("保存数据[%s]至时序数据库失败: %s", data.toString(), e.getMessage()), e);
        }
    }

    public void write(String database, String clientId, DataType type, JSONArray dataArray) {
        try {
            BatchPoints batchPoints = BatchPoints.database(database).tag("async", "true").
                    retentionPolicy("default").consistency(ConsistencyLevel.ALL).build();
            for (Iterator<Object> iterator = dataArray.iterator();iterator.hasNext();) {
                JSONObject data = (JSONObject) iterator.next();
                String measurement = type.toString();
                Map<String, String> tags = new HashMap<String, String>();
                tags.put("clientId", clientId);

                Map fields = (Map) data;
                Long timestamp = data.getLong("time");
                fields.remove("time");

                Point point = Point.measurement(measurement)
                        .time(timestamp, TimeUnit.MILLISECONDS).tag(tags).fields(fields).build();
                batchPoints.point(point);
            }

            influxDB.write(batchPoints);
        } catch (Exception e) {
            LOGGER.error(String.format("保存数据[%s]至时序数据库失败: %s", dataArray.toString(), e.getMessage()), e);
        }
    }

    public String query(String database, String clientId, DataType type, Date start, Date end, int limit, int count) {
        String queryParam = null;
        try {
            queryParam = String.format("select * from %s where clientId = %s and time >= start and time < end limit %d count %d",
                type.toString(), clientId, start.getTime(), end.getTime(), limit, count);
            Query query = new Query(queryParam, database);
            QueryResult result = influxDB.query(query);

            return result.getResults().toString();
        } catch (Exception e) {
            LOGGER.error(String.format("时序数据查询[%s]失败: %s", queryParam, e.getMessage()), e);
        }

        return null;
    }
}
